package com.wuwangfu.func;

import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.LocalTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Description：全窗口函数
 * @Author：jcshen
 * @Date：2023-07-06
 */
public class UrlsProcessWindowFunc01 extends ProcessWindowFunction<HashMap<String,Long>,String,Boolean, TimeWindow> {

    /**
     *
     * @param key 分区key返回的值
     * @param cxt 上下文（可获取当前处理时间，水位线，窗口状态）
     * @param iterable 全量数据的迭代器
     * @param out
     * @throws Exception
     */
    @Override
    public void process(Boolean key,Context cxt, Iterable<HashMap<String, Long>> iterable, Collector<String> out) throws Exception {
        ArrayList<HashMap<String, Long>> list = new ArrayList<>();

        for (HashMap<String, Long> map : iterable){
            list.add(map);
        }
        //这里数据是全部的，所以迭代器直接获取第一个数据就行
        HashMap<String, Long> values = list.get(0);
        List<Map.Entry<String, Long>> finalValue = sortByMap(values);

        TimeWindow window = cxt.window();
        long start = window.getStart();
        long end = window.getEnd();

        out.collect("当前时间：" + LocalTime.now() + "分区key："+ key + "窗口起始时间："+
                new Date(start) + "\t 窗口结束时间：" + new Date(end) + "\t 排在前五名的是："
                + finalValue.toString());
    }

    private List<Map.Entry<String,Long>> sortByMap(HashMap<String, Long> values){
        return values.entrySet()
                .stream()
                .sorted(Map.Entry.comparingByValue(Comparator.reverseOrder()))
                .collect(Collectors.toList());
    }
}
